package org.idea.qiyu.framework.canal.handle;

import com.alibaba.otter.canal.protocol.Message;
import org.idea.qiyu.framework.canal.producer.CanalMessageProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.concurrent.ArrayBlockingQueue;


/**
 * 消费队列的处理
 *
 * @Author linhao
 * @Date created in 3:30 下午 2021/6/20
 */
@Component
public class CanalMessageQueueHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(CanalMessageQueueHandler.class);

    /**
     * 内存堆积消息上限氏1500，超过会堵塞 线程安全队列
     */
    private static ArrayBlockingQueue<Message> arrayBlockingQueue = new ArrayBlockingQueue(1500);

    /**
     * 非堵塞添加，超过容量回导致消息丢失
     *
     * @param message
     */
    public void addMessage(Message message) {
        try {
            arrayBlockingQueue.add(message);
        } catch (Exception e) {
            LOGGER.error("[arrayBlockingQueue] e is {}", e);
        }
    }

    /**
     * 非堵塞
     *
     * @return
     */
    public Message takeMessage() {
        return arrayBlockingQueue.poll();
    }

}
